package cn.tpshion.devops.service.impl;

import cn.hutool.core.date.BetweenFormatter;
import cn.hutool.core.date.DateUtil;
import cn.tpshion.devops.common.constant.NodeConstant;
import cn.tpshion.devops.common.enums.MsgType;
import cn.tpshion.devops.common.enums.NodeStatus;
import cn.tpshion.devops.common.CoreException;
import cn.tpshion.devops.common.constant.RedisKey;
import cn.tpshion.devops.domain.dto.TaskStatusDTO;
import cn.tpshion.devops.domain.dto.TaskWorkDTO;
import cn.tpshion.devops.domain.entity.*;
import cn.tpshion.devops.handle.SSEHandle;
import cn.tpshion.devops.node.Node;
import cn.tpshion.devops.node.support.NodeContext;
import cn.tpshion.devops.node.support.NodeFactory;
import cn.tpshion.devops.node.support.NodeTask;
import cn.tpshion.devops.service.NodeService;
import cn.tpshion.devops.util.AssertUtil;
import com.alibaba.fastjson2.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.FutureTask;

@Slf4j
@Service
public class NodeServiceImpl implements NodeService {

    @Resource
    private ThreadPoolTaskExecutor taskExecutor;

    @Resource
    private NodeFactory nodeFactory;

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    private Map<String, NodeTask> taskMap = new ConcurrentHashMap<>();

    @Resource
    private SSEHandle sseHandle;

    @Resource
    private RedissonClient redissonClient;

    @Override
    public void start(TaskWorkDTO dto) {
        RLock lock = redissonClient.getLock(RedisKey.TASK_LOCK + dto.getTaskId());
        AssertUtil.isTrue(!lock.tryLock(), "任务处理中，请等待任务结束");
        try {
            AssertUtil.isTrue(taskMap.containsKey(dto.getTaskId()), "任务处理中，请等待任务结束");

            HashOperations<String, String, String> hash = redisTemplate.opsForHash();
            String value = hash.get(RedisKey.TASK_KEY, dto.getTaskId());
            AssertUtil.isTrue(StringUtils.isBlank(value), "数据异常");

            TaskWork taskWork = JSON.parseObject(value, TaskWork.class);
            AssertUtil.isEmpty(taskWork, "数据异常");
            AssertUtil.isEmpty(taskWork.getNodeDataList(), "流程节点数据异常");

            List<NodeData> nodeDataList = taskWork.getNodeDataList();

            Long batchId = getBatchId(taskWork.getTaskId());

            NodeContext nodeContext = new NodeContext();
            nodeContext.setTaskId(dto.getTaskId());
            nodeContext.setTaskWork(taskWork);
            nodeContext.setBatchId(batchId);

            FutureTask task = new FutureTask(() -> {
                log.info("task start");
                beforeFlow(nodeContext);

                sseHandle.sendMessage(nodeContext, new MsgData(MsgType.SUCCESS, "开始执行节点任务"));

                boolean interrupted = false;
                for (NodeData nodeData : nodeDataList) {
                    if(interrupted){
                        break;
                    }
                    if(Thread.currentThread().isInterrupted()){
                        interrupted = true;
                        break;
                    }
                    Node node = nodeFactory.getNode(nodeData.getType());
                    try {
                        nodeContext.setNodeId(nodeData.getId());
                        nodeContext.setNodeData(nodeData);

                        // 前置处理
                        beforeHandle(nodeContext);

                        // 处理任务
                        node.handle(nodeContext);

                        // 后置处理
                        afterHandle(nodeContext);
                    } catch (Exception e) {
                        if(e instanceof InterruptedException){
                            // ignore
                            interrupted = true;
                        }else{
                            log.error("startTask node error:{}", e.getMessage(), e);
                            // 异常处理
                            errorHandle(nodeContext, e);
                            taskMap.remove(nodeContext.getTaskId());
                            throw new CoreException("任务失败");
                        }
                    }
                }
                if(!interrupted){
                    log.info("task success");
                    sseHandle.sendMessage(nodeContext, new MsgData(MsgType.SUCCESS, "执行节点任务成功"));
                    afterFlow(nodeContext);
                }else{
                    saveFlowHistory(nodeContext, NodeStatus.STOP);
                }
                taskMap.remove(nodeContext.getTaskId());
                return true;
            });

            NodeTask nodeTask = new NodeTask();
            nodeTask.setTaskId(dto.getTaskId());
            nodeTask.setTaskWork(taskWork);
            nodeTask.setFutureTask(task);

            taskMap.put(dto.getTaskId(), nodeTask);
            taskExecutor.submit(task);
        }finally {
            if(lock.isHeldByCurrentThread()){
                lock.unlock();
            }
        }
    }

    private Long getBatchId(String taskId){
        HashOperations<String, String, Long> hash = redisTemplate.opsForHash();
        return hash.increment(RedisKey.HISTORY_ID, taskId, 1);
    }

    private void beforeFlow(NodeContext nodeContext){
        FlowStatusData status = new FlowStatusData();
        status.setTaskId(nodeContext.getTaskId());
        status.setStatus(true);

        MsgData<FlowStatusData> data = new MsgData<>(MsgType.FLOW);
        data.setData(status);
        sseHandle.sendMessage(nodeContext, data);
    }

    private void afterFlow(NodeContext nodeContext){
        FlowStatusData status = new FlowStatusData();
        status.setTaskId(nodeContext.getTaskId());
        status.setStatus(false);

        MsgData<FlowStatusData> data = new MsgData<>(MsgType.FLOW);
        data.setData(status);
        sseHandle.sendMessage(nodeContext, data);

        saveFlowHistory(nodeContext, NodeStatus.SUCCESS);
    }

    private void saveFlowHistory(NodeContext nodeContext, NodeStatus status){
        FlowHistory history = new FlowHistory();
        history.setTaskId(nodeContext.getTaskId());
        history.setBatchId(nodeContext.getBatchId());
        history.setStartTime(nodeContext.getCreateTime().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
        LocalDateTime endTime = LocalDateTime.now();
        history.setEndTime(endTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
        history.setTotalTime(DateUtil.formatBetween(Duration.between(nodeContext.getCreateTime(), endTime).toMillis(), BetweenFormatter.Level.SECOND));
        history.setStatus(status.getStatus());
        history.setTitle(nodeContext.getTaskWork().getName());

        Map<String, Object> contextData = nodeContext.getData();
        Object commitMsg = contextData.get(NodeConstant.GIT_COMMIT_MSG);
        history.setCommitMsg(Objects.nonNull(commitMsg) ? commitMsg.toString() : "");

        Object commitVersion = contextData.get(NodeConstant.GIT_COMMIT_VERSION);
        history.setCommitVersion(Objects.nonNull(commitVersion) ? commitVersion.toString() : "");

        Object branch = contextData.get(NodeConstant.GIT_BRANCH);
        history.setBranch(Objects.nonNull(branch) ? branch.toString() : "");

        HashOperations<String, String, String> hash = redisTemplate.opsForHash();
        hash.put(RedisKey.HISTORY + nodeContext.getTaskId(), nodeContext.getBatchId().toString(), JSON.toJSONString(history));
    }

    private void beforeHandle(NodeContext nodeContext){
        NodeData nodeData = nodeContext.getNodeData();
        nodeData.setStatus(NodeStatus.RUN.getStatus());

        handleNodeStatus(nodeContext, nodeData);

        sseHandle.sendMessage(nodeContext, new MsgData(MsgType.SUCCESS, "开始执行：" + nodeData.getTitle()));
    }

    private void afterHandle(NodeContext nodeContext){
        NodeData nodeData = nodeContext.getNodeData();
        nodeData.setStatus(NodeStatus.SUCCESS.getStatus());

        handleNodeStatus(nodeContext, nodeData);

        sseHandle.sendMessage(nodeContext, new MsgData(MsgType.SUCCESS, nodeData.getTitle() + " 执行成功"));
    }

    private void errorHandle(NodeContext nodeContext, Exception e){
        NodeData nodeData = nodeContext.getNodeData();
        nodeData.setStatus(NodeStatus.ERROR.getStatus());

        handleNodeStatus(nodeContext, nodeData);

        sseHandle.sendMessage(nodeContext, new MsgData(MsgType.ERROR, nodeData.getTitle() + " 执行失败：" + e.getMessage()));
        StringBuilder builder = new StringBuilder();
        for (StackTraceElement traceElement : e.getStackTrace()) {
            builder.append(traceElement.toString());
        }
        sseHandle.sendMessage(nodeContext, new MsgData(MsgType.ERROR, builder.toString()));

        saveFlowHistory(nodeContext, NodeStatus.ERROR);
    }

    private void handleNodeStatus(NodeContext nodeContext, NodeData nodeData){
        NodeStatusData status = new NodeStatusData();
        status.setId(nodeData.getId());
        status.setStatus(nodeData.getStatus());

        MsgData<NodeStatusData> data = new MsgData<>(MsgType.NODE);
        data.setData(status);
        sseHandle.sendMessage(nodeContext, data);
    }

    @Override
    public void stopTask(TaskWorkDTO dto) {
        NodeTask nodeTask = taskMap.get(dto.getTaskId());
        if(Objects.isNull(nodeTask)){
            return;
        }

        FutureTask task = nodeTask.getFutureTask();
        if(Objects.nonNull(task)){
            task.cancel(true);
        }
    }

    @Override
    public TaskStatusDTO taskStatus(TaskWorkDTO dto) {
        TaskStatusDTO taskStatus = new TaskStatusDTO();
        // 先获取运行中任务
        NodeTask nodeTask = taskMap.get(dto.getTaskId());
        if(Objects.nonNull(nodeTask) && Objects.nonNull(nodeTask.getTaskWork())){
            taskStatus.setTaskWork(nodeTask.getTaskWork());
            taskStatus.setRunning(true);
            return taskStatus;
        }

        HashOperations<String, String, String> hash = redisTemplate.opsForHash();
        String value = hash.get(RedisKey.TASK_KEY, dto.getTaskId());
        TaskWork taskWork = JSON.parseObject(value, TaskWork.class);
        taskStatus.setTaskWork(taskWork);
        taskStatus.setRunning(false);
        return taskStatus;
    }

    @Override
    public Boolean runStatus(TaskWorkDTO dto) {
        NodeTask nodeTask = taskMap.get(dto.getTaskId());
        return Objects.nonNull(nodeTask) && Objects.nonNull(nodeTask.getTaskWork());
    }
}
